/**
 * Copyright (C) 2024-2030 Super BilGen & 超亿时代信息技术股份有限公司.All Rights Reserved.
 */
package cn.com.yang.file;

/**
 * @Model cn.com.yang.thread.MultiThreadFileReader
 * @Desc  TODO
 * @author yangyijin
 * @since 2025年5月22日-下午8:25:11
 */
import java.io.*;
import java.util.concurrent.*;
import java.util.*;

public class MultiThreadFileReader {

    private static final int NUM_THREADS = 4;  // 使用4个线程
    private static final String FILE_PATH = "F:\\20gb_data.txt";  // 文件路径

    public static void main(String[] args) throws InterruptedException, ExecutionException, IOException {
        // 创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(NUM_THREADS);

        // 获取文件大小
        File file = new File(FILE_PATH);
        long fileSize = file.length();
        long chunkSize = fileSize / NUM_THREADS;

        List<Future<Void>> futures = new ArrayList<>();

        for (int i = 0; i < NUM_THREADS; i++) {
            // 每个线程处理文件的一部分
            long startPosition = i * chunkSize;
            long endPosition = (i == NUM_THREADS - 1) ? fileSize : (i + 1) * chunkSize;

            // 确保每个线程从完整的行开始
            if (startPosition > 0) {
                startPosition = findNextLineStart(startPosition);
            }

            futures.add(executorService.submit(new FileReadTask(startPosition, endPosition)));
        }

        // 等待所有线程完成
        for (Future<Void> future : futures) {
            future.get();
        }

        executorService.shutdown();
    }

    // 找到下一个行的起始位置
    private static long findNextLineStart(long startPosition) throws IOException {
        try (RandomAccessFile randomAccessFile = new RandomAccessFile(FILE_PATH, "r")) {
            randomAccessFile.seek(startPosition);
            while (randomAccessFile.getFilePointer() < randomAccessFile.length()) {
                int byteRead = randomAccessFile.read();
                if (byteRead == '\n') {
                    return randomAccessFile.getFilePointer();
                }
            }
        }
        return startPosition;
    }

    // 文件读取任务
    static class FileReadTask implements Callable<Void> {
        private long startPosition;
        private long endPosition;

        public FileReadTask(long startPosition, long endPosition) {
            this.startPosition = startPosition;
            this.endPosition = endPosition;
        }

        @Override
        public Void call() throws Exception {
            try (RandomAccessFile randomAccessFile = new RandomAccessFile(FILE_PATH, "r")) {
                // 定位到读取的起始位置
                randomAccessFile.seek(startPosition);

                // 使用UTF-8编码读取
                InputStreamReader inputStreamReader = new InputStreamReader(new FileInputStream(randomAccessFile.getFD()), "UTF-8");
                BufferedReader bufferedReader = new BufferedReader(inputStreamReader);

                String line;
                while (randomAccessFile.getFilePointer() < endPosition && (line = bufferedReader.readLine()) != null) {
                    // 处理每一行
                    processLine(line);
                }

                // 检查是否需要读取最后一行（如果文件没有换行符结尾）
                if (randomAccessFile.getFilePointer() < randomAccessFile.length()) {
                    String lastLine = bufferedReader.readLine();
                    if (lastLine != null) {
                        processLine(lastLine);
                    }
                }

            }
            return null;
        }

        private void processLine(String line) {
            // 处理每一行的数据
            System.out.println(Thread.currentThread().getName() + " is processing: " + line);
        }
    }
}

